package pipelines;/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import fragment.IcebergDeletedSink;
import fragment.IcebergSink;
import fragment.TchouseDeletedSink;
import fragment.TchouseSink;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

public class JobUtils {

    private static final Logger LOGGER = LoggerFactory.getLogger(JobUtils.class);

    public static void addAllSinks(String sourceName, ParameterTool pt, TableEnvironment tableEnv) {
        List<String> sinks = new ArrayList<>();
        List<String> deletedSinks = new ArrayList<>();
        String[] activeSinks = pt.get("ACTIVE_SINKS").split(",");
        for (String activeSink : activeSinks) {
            if (activeSink.equalsIgnoreCase("iceberg")) {
                sinks.add(new IcebergSink().createConnector(pt, tableEnv));
            } else if (activeSink.equalsIgnoreCase("iceberg_delete")) {
                deletedSinks.add(new IcebergDeletedSink().createConnector(pt, tableEnv));
            } else if (activeSink.equalsIgnoreCase("tchouse")) {
                sinks.add(new TchouseSink().createConnector(pt, tableEnv));
            } else if (activeSink.equalsIgnoreCase("tchouse_delete")) {
                deletedSinks.add(new TchouseDeletedSink().createConnector(pt, tableEnv));
            }
        }

        StatementSet statementSet = tableEnv.createStatementSet();
        for (String sink : sinks) {
            String insertSql = String.format("insert into %s select * from %s",
                    sink, sourceName);
            LOGGER.info("insert as\n {}", insertSql);
            statementSet.addInsertSql(insertSql);
        }

        for (String deletedSink : deletedSinks) {
            String insertDeletedSql = String.format("insert into %s select * from %s where type not in (0, 2, 3)",
                    deletedSink, sourceName);
            LOGGER.info("insert deleted as\n {}", insertDeletedSql);
            statementSet.addInsertSql(insertDeletedSql);
        }
        TableResult tableResult2 = statementSet.execute();
        System.out.println(tableResult2.getJobClient().get().getJobStatus());
    }

}
